package com.kafka_demo2;

import com.alibaba.fastjson2.JSON;
import com.kafka_demo.Order;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class MsgProducer {
    private final static String TOPIC_NAME = "my-replicated-topic";

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node01:9092,node02:9092,node03:9092");

        // 设置key，value序列化方式
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        int msgNum = 5;
        for (int i = 1; i <= msgNum; i++) {
            Order order = new Order(i, 100 + i, 1, 1000.00);
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME,
                    order.getOrderId().toString(), JSON.toJSONString(order));

            // 同步发送消息
//            Future<RecordMetadata> future = producer.send(record);
//            RecordMetadata metadata = future.get();

            // 异步发送
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.out.println("发送消息异常:" + exception.getMessage());
                        return;
                    }
                    System.out.println(metadata.topic() + "---" + metadata.partition() + "---" + metadata.offset());
                }
            });
        }

        Thread.sleep(1000);

    }
}